package cn.monkey.transport.mq.redisson;

import cn.money.transport.mq.MapExchangeContext;
import cn.monkey.transport.core.Exchange;
import cn.monkey.transport.core.ExchangeContext;
import io.netty.util.AttributeKey;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public class SimpleRedissonExchange implements RedissonExchange {
    private final String id;
    private final RedissonInbound inbound;
    private final RedissonOutbound outbound;
    private final AtomicBoolean closed;
    private final MapExchangeContext context;

    public SimpleRedissonExchange(String id, RedissonInbound inbound, RedissonOutbound outbound) {
        this.id = id;
        this.inbound = inbound;
        this.outbound = outbound;
        this.closed = new AtomicBoolean(false);
        this.context = new MapExchangeContext();
    }

    @Override
    public String id() {
        return this.id;
    }

    public ExchangeContext context() {
        return this.context;
    }

    @Override
    public Mono<Void> close() {
        this.closed.set(true);
        List<Mono<Void>> list = List.of(this.inbound().close(), this.outbound().close());
        return Flux.merge(list).then();
    }

    @Override
    public boolean isActive() {
        return !this.closed.get();
    }

    @Override
    public RedissonInbound inbound() {
        return this.inbound;
    }

    @Override
    public RedissonOutbound outbound() {
        return this.outbound;
    }

    @Override
    public RedissonExchange bind() {
        this.setAttribute(AttributeKey.valueOf(Exchange.class, Exchange.KEY).name(), this);
        return this;
    }

    @Override
    public RedissonExchange inverse() {
        SimpleRedissonInbound inbound = new SimpleRedissonInbound(this.outbound().topic(), this.outbound().redissonClient(), this.outbound().codec(), this.inbound().pollSize(), this.inbound().pollFrequencyMs());
        SimpleRedissonOutbound outbound = new SimpleRedissonOutbound(this.inbound().topic(), this.inbound().redissonClient(), this.inbound().codec(), this.outbound().cacheSize());
        return new SimpleRedissonExchange(this.id(), inbound, outbound);
    }
}
